/**
 * @file crop_ss_par_seq.cpp
 * @author Andreas Bok Andersen
 * @brief CRoP SpaceSaving with parallel read and sequential reduction
 */

#include "crop.cpp"
#include <stdio.h>
#include <mpi.h>
#define MPI_H
#include <string.h>
#include "fileparser.cpp"
#include <time.h>

template<class T, class U>
void run(int argc, char *argv[]) {
	int rank, comm_size, k = 4;
	string FILENAME;
	double start, end, split;

	MPI_Init(&argc, &argv);
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &comm_size);

	CmdParser cp;
	if (!cp.parseargs(argc, argv)) {
		MPI_Abort(MPI_COMM_WORLD, -1);

	}
	if (rank == 0) {
		cp.printargs();
		cout << cp.get_cmdargs() << endl;
	}
	FILENAME = DATADIR + cp.a.get < string > (INFILE);
	const char * FILEPATH = FILENAME.c_str();
	k = cp.a.get<int>(KSIZE);
	MPI_Status status;

	MPI_Datatype entrytype;
	ENTRY_TYPE(&entrytype);

	Crop<T> *crop = new Crop<T>(rank, comm_size, k);
	start = MPI_Wtime();
	MMapFILEParser<U> fp((char*) FILEPATH); // Change to command arg

	while (fp.getinput()) {
		crop->crop((int*) fp.CURR_PTR, fp.NUM_PTR, fp.count);
	}
	end = MPI_Wtime();
	fprintf(stdout, "%d\t%d\tCROP\t%.8f\n", comm_size, rank, end - start);

	bool done = false;

	int summarysize = crop->summarysize();
	Entry * entries = (Entry*) malloc(k * sizeof(Entry));
	//fprintf(stderr, "rank: %d, %s\n", rank, "crop_ss.getting summary");
	crop->get_summary(entries);

	fprintf(stderr, "rank: %d %s #%d\n", rank, "crop_ss.got_summary", summarysize);

	Entry *resv_buf = NULL;
	if (rank == 0) {
		resv_buf = (Entry*) malloc(sizeof(Entry) * k * comm_size);
		//printf("Sizeof resvbuf: %d\n", sizeof(resv_buf));
	}
	split = MPI_Wtime();
	MPI_Gather(entries, summarysize, entrytype, resv_buf, k, entrytype, 0,
			MPI_COMM_WORLD);

	fprintf(stderr,"%s %d\n", "Called gather at ", rank);

	if (rank == 0) {
		HASH_ENTRY_REF_MAP sum;
		fprintf(stderr, "%s %d\n", "Finishing at ", rank);
		for (int i = 0; i < (k * comm_size); i++) {
			if (resv_buf[i].i == -1)
				continue;
			Entry *e;
			if (sum.find(resv_buf[i].hash) == sum.end()) {
				sum.insert(
						make_pair<size_t, Entry*>(resv_buf[i].hash,
								&resv_buf[i]));
			} else {
				Entry *e = sum.find(resv_buf[i].hash)->second;
				e->weight += resv_buf[i].weight;
				e->eps += resv_buf[i].eps;
			}
		}
		HASH_ENTRY_REF_MAP::iterator it;
		ENTRY_VEC entryvec;
		for (it = sum.begin(); it != sum.end(); it++) {
			entryvec.push_back(*it->second);
		}
		sort(entryvec.begin(), entryvec.end(), sort_reverse);
		entryvec.resize(k);
		stringstream ss;
		ss << FILENAME << "_k" << k << ".res";
		FILENAME = ss.str();
		writetofile((char*) FILENAME.c_str(), entryvec);
		free(resv_buf);
	}
	end = MPI_Wtime();
	fprintf(stdout, "%d\t%d\tGAT\t%.8f\n", comm_size, rank, end - split);
	fprintf(stdout, "%d\t%d\tTOT\t%.8f\n", comm_size, rank, end - start);
	delete entries;
	MPI_Type_free(&entrytype);
	delete crop;
	MPI_Finalize();
	//return 0;
}

int main(int argc, char *argv[]) {
	CmdParser cp;
	if (!cp.parseargs(argc, argv)) {
		return -1;
	}

	if (cp.a.get < string > (FORMAT) == "dcs") {
		if (cp.a.get < string > (STYPE) == "buf") {
			run<StreamSummaryBuffer,DCS_FORMAT>(argc, argv);
		} else {
			run<StreamSummaryBucket, DCS_FORMAT>(argc, argv);
		}
	} else if (cp.a.get < string > (FORMAT) == "coo"){
		if (cp.a.get < string > (STYPE) == "buf") {
			run<StreamSummaryBuffer,COO_FORMAT>(argc, argv);
		} else {
			run<StreamSummaryBucket,COO_FORMAT>(argc, argv);
		}
	}
}
